package com.es.controller;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.util.StrUtil;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.es.dto.HotelData;
import com.es.dto.HotelDto;
import com.es.dto.PriceAgg;
import com.es.entity.Hotel;
import com.es.service.IHotelService;
import com.es.utils.ElasticUtil;
import com.es.utils.HttpStatus;
import com.es.utils.KafkaUtil;
import com.es.utils.Result;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.geo.GeoPoint;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.functionscore.ScriptScoreFunctionBuilder;
import org.elasticsearch.index.query.functionscore.ScriptScoreQueryBuilder;
import org.elasticsearch.script.Script;
import org.elasticsearch.script.ScriptType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.range.GeoDistanceAggregationBuilder;
import org.elasticsearch.search.aggregations.bucket.range.ParsedGeoDistance;
import org.elasticsearch.search.aggregations.bucket.range.Range;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.Stats;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.GeoDistanceSortBuilder;
import org.elasticsearch.search.sort.SortBuilders;
import org.elasticsearch.search.sort.SortOrder;
import org.elasticsearch.search.suggest.Suggest;
import org.elasticsearch.search.suggest.SuggestBuilder;
import org.elasticsearch.search.suggest.SuggestBuilders;
import org.elasticsearch.search.suggest.completion.CompletionSuggestion;
import org.elasticsearch.search.suggest.completion.CompletionSuggestionBuilder;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.util.StringUtils;
import org.springframework.web.bind.annotation.*;

import java.io.IOException;
import java.util.*;

@RestController
@RequestMapping("/hotel")
@Slf4j
public class HotelController {
    @Autowired
    private IHotelService hotelService;

    @Autowired
    private ElasticUtil elasticUtil;

    @Autowired
    private KafkaUtil kafkaUtil;

    @Value("${hotel.topic}")
    private String topicName;

    private static final String INDEX_NAME = "t_hotel";

    private static final String SUGGEST_INDEX_NAME = "t_hotel_suggest";

    private static final String chineseSugName = "hotel_chinese_sug";
    private static final String fullPinyinSugName = "hotel_full_pinyin_sug";
    private static final String headPinyinSugName = "hotel_head_pinyin_sug";

    private static final String PRICE_AGG = "price_agg";
    private static final String STAR_AGG = "star_agg";
    private static final String GEO_AGG = "geo_agg";

    @PostMapping("/addHotel")
    public Result<Object> addHotel(@RequestBody HotelDto hotelDto) {
        String message = null;
        Hotel one = hotelService.getOne(new QueryWrapper<Hotel>().eq("id", hotelDto.getId()));
        if (one != null) {
            message = "主键已经存在，请重新添加";
        } else {
            Hotel hotel = new Hotel();
            BeanUtils.copyProperties(hotelDto, hotel);
            if (hotelService.save(hotel)) {
                message = "保存成功";
            } else {
                message = "保存失败";
            }
        }
        return Result.builder().code(HttpStatus.OK.getCode()).data(message).build();
    }

    @GetMapping("/del/{hotelId}")
    public Result<Object> delHotel(@PathVariable("hotelId") Long hotelId) {
        Hotel one = hotelService.getOne(new QueryWrapper<Hotel>().eq("id", hotelId));
        String message = "";
        if (one == null) {
            message = "没有找到相关记录";
        } else {
            hotelService.removeById(hotelId);
            message = "删除成功";
        }
        return Result.builder().code(HttpStatus.OK.getCode()).data(message).build();
    }

    @GetMapping("/get/{hotelId}")
    public Result<Object> getHotel(@PathVariable("hotelId") Long hotelId) {
        Hotel one = hotelService.getOne(new QueryWrapper<Hotel>().eq("id", hotelId));
        Object message = "";
        if (one == null) {
            message = "没有找到相关记录";
        } else {
            message = one;
        }
        return Result.builder().code(HttpStatus.OK.getCode()).data(message).build();
    }

    @PostMapping("/saveHotel")
    public Result<Object> saveHotel(@RequestBody HotelDto hotelDto) {
        String message = "";
        Hotel one = hotelService.getOne(new QueryWrapper<Hotel>().eq("id", hotelDto.getId()));
        if (one == null) {
            message = "更新记录不存在";
        } else {
            Hotel hotel = new Hotel();
            BeanUtils.copyProperties(hotelDto, hotel);
            if (hotelService.updateById(hotel)) {
                message = "保存成功";
            } else {
                message = "保存失败";
            }
        }
        return Result.builder().code(HttpStatus.OK.getCode()).data(message).build();
    }

    @PostMapping("/list")
    public Result<Object> list(@RequestBody HotelData hotelData) {
        QueryWrapper<Hotel> wrapper = new QueryWrapper<>();
        if (hotelData.getQuery() != null) {
            HotelDto query = hotelData.getQuery();
            if (!StringUtils.isEmpty(query.getTitle())) {
                wrapper.like("title", query.getTitle());
            }
        }
        IPage<Hotel> pageList = hotelService.page(new Page<>(hotelData.getFrom(), hotelData.getSize()), wrapper);

        HotelData resData = HotelData.builder().build();

        List<HotelDto> dtoList = new ArrayList<>();
        List<Hotel> hotelList = pageList.getRecords();
        hotelList.forEach(item -> {
            HotelDto hotelDto = new HotelDto();
            BeanUtils.copyProperties(item, hotelDto);
            dtoList.add(hotelDto);
        });
        resData.setData(dtoList);
        resData.setTotal(pageList.getTotal());
        return Result.builder()
                .code(HttpStatus.OK.getCode())
                .data(resData).build();
    }

    @PostMapping("/allcity")
    public Result<Object> allcity() throws IOException {
        RestHighLevelClient highLevelClient = elasticUtil.getHighLevelClient();
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        TermsAggregationBuilder aggregation = AggregationBuilders.terms("city_group").field("city");
        searchSourceBuilder.aggregation(aggregation);
        SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
        searchRequest.source(searchSourceBuilder);

        SearchResponse response = highLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        Aggregations aggregations = response.getAggregations();
        Terms city_group = aggregations.get("city_group");
        List<? extends Terms.Bucket> buckets = city_group.getBuckets();
        List<String> cityes = new ArrayList<>();
        buckets.stream().forEach(item -> {
            cityes.add(item.getKeyAsString());
        });
        return Result.builder()
                .code(HttpStatus.OK.getCode())
                .data(cityes).build();
    }

    @PostMapping("/search")
    public Result<Object> search(@RequestBody HotelData hotelData) throws IOException {
        HotelData resData = HotelData.builder().build();
        SearchRequest searchRequest = new SearchRequest(INDEX_NAME);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        sourceBuilder.from(hotelData.getFrom());
        sourceBuilder.size(hotelData.getSize());
        BoolQueryBuilder builder = getFilter(hotelData);
        ScriptScoreQueryBuilder scriptQueryBuilder = QueryBuilders.scriptScoreQuery(builder, new ScriptScoreFunctionBuilder(getScript(hotelData)));
        sourceBuilder.query(scriptQueryBuilder);
        // 处理聚合
        List<AggregationBuilder> builderList = getAgg(hotelData);
        builderList.forEach(item -> {
            sourceBuilder.aggregation(item);
        });
        if (hotelData.getQuery() != null) {
            String sort = hotelData.getQuery().getSort();
            if (StrUtil.isNotBlank(sort)) {
                String sortName = sort.split(" ")[0];
                String sortType = sort.split(" ")[1];
                if (sortName.equals("location")) {
                    double lat = hotelData.getQuery().getLat();
                    double lon = hotelData.getQuery().getLon();
                    GeoDistanceSortBuilder sortBuilder = SortBuilders.geoDistanceSort("location", lat, lon)
                            .point(lat, lon).unit(DistanceUnit.KILOMETERS);
                    sourceBuilder.sort(sortBuilder);
                } else {
                    sourceBuilder.sort(sortName, sortType.equals("asc") ? SortOrder.ASC : SortOrder.DESC);
                }
            }
        }
        searchRequest.source(sourceBuilder);
        log.info("Query DSL : {}", sourceBuilder.toString());
        RestHighLevelClient client = elasticUtil.getHighLevelClient();
        SearchResponse response = client.search(searchRequest, RequestOptions.DEFAULT);
        SearchHits hits = response.getHits();
        SearchHit[] hitsHits = hits.getHits();
        resData.setTotal(hits.getTotalHits().value);
        for (SearchHit hit : hitsHits) {
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();
            HotelDto hotelDto = new HotelDto();
            if (sourceAsMap.get("city") != null) {
                hotelDto.setCity(StrUtil.join(",", (List) sourceAsMap.get("city")));
            }
            hotelDto.setId((Long) sourceAsMap.get("id"));
            hotelDto.setAddress((String) sourceAsMap.get("address"));
            hotelDto.setTitle((String) sourceAsMap.get("title"));
            hotelDto.setFavourablePercent((Integer) sourceAsMap.get("favourable_percent"));
            if (sourceAsMap.get("location") != null) {
                Map<String, Object> location = (Map<String, Object>) sourceAsMap.get("location");
                hotelDto.setLat((Double) location.get("lat"));
                hotelDto.setLon((Double) location.get("lon"));
            }
            if (sourceAsMap.get("star") != null) {
                hotelDto.setStar(StrUtil.join(",", (List) sourceAsMap.get("star")));
            }
            if (sourceAsMap.get("impression") != null) {
                hotelDto.setImpression(StrUtil.join(",", (List) sourceAsMap.get("impression")));
            }
            if (sourceAsMap.get("business_district") != null) {
                hotelDto.setBusinessDistrict(StrUtil.join(",", (List) sourceAsMap.get("business_district")));
            }
            hotelDto.setFullRoom((Boolean) sourceAsMap.get("full_room"));
            hotelDto.setPrice(String.valueOf(sourceAsMap.get("price")));
            if (sourceAsMap.get("pic") != null) {
                hotelDto.setPic(StrUtil.join(",", (List) sourceAsMap.get("pic")));
            }
            if (resData.getData() == null) {
                resData.setData(new ArrayList<>());
            }
            resData.getData().add(hotelDto);
        }
        return Result.builder()
                .code(HttpStatus.OK.getCode())
                .data(resData).build();
    }

    @PostMapping("/sugsearch")
    public Result<Object> suggestSearch(@RequestBody Map<String, String> params) throws Exception {
        String prefixWord = params.get("prefix");
        if (StrUtil.isBlank(prefixWord)) {
            return Result.builder()
                    .code(HttpStatus.OK.getCode())
                    .data(ListUtil.empty()).build();
        }

        SearchRequest searchRequest = new SearchRequest(SUGGEST_INDEX_NAME);
        SearchSourceBuilder builder = new SearchSourceBuilder();
        SuggestBuilder suggestBuilder = new SuggestBuilder();
        CompletionSuggestionBuilder chineseBuilder = SuggestBuilders.completionSuggestion("chinese").prefix(prefixWord);
        suggestBuilder.addSuggestion(chineseSugName, chineseBuilder);

        CompletionSuggestionBuilder fullPinyinBuilder = SuggestBuilders.completionSuggestion("full_pinyin").prefix(prefixWord);
        suggestBuilder.addSuggestion(fullPinyinSugName, fullPinyinBuilder);

        CompletionSuggestionBuilder headPinyinBuilder = SuggestBuilders.completionSuggestion("head_pinyin").prefix(prefixWord);
        suggestBuilder.addSuggestion(headPinyinSugName, headPinyinBuilder);
        builder.suggest(suggestBuilder);
        searchRequest.source(builder);

        RestHighLevelClient highLevelClient = elasticUtil.getHighLevelClient();
        SearchResponse searchResponse = highLevelClient.search(searchRequest, RequestOptions.DEFAULT);
        Suggest suggest = searchResponse.getSuggest();
        HashSet sugSet = new HashSet();
        CompletionSuggestion chineseSuggestion = suggest.getSuggestion(chineseSugName);
        List<String> sugList = new ArrayList<String>();
        //遍历中文suggest结果
        for (CompletionSuggestion.Entry.Option option : chineseSuggestion.getOptions()) {
            Map<String, Object> sourceMap = option.getHit().getSourceAsMap();//获取文档
            Map<String, Object> chineseText = (Map<String, Object>) sourceMap.get("chinese");//获取文档中的中文
            String sugChinese = (String) chineseText.get("input");
            if (!sugSet.contains(sugChinese)) {
                sugList.add(sugChinese);
                sugSet.add(sugChinese);
            }
        }

        CompletionSuggestion fullPinyinSuggestion = suggest.getSuggestion(fullPinyinSugName);//获取suggest结果
        //遍历拼音全拼suggest结果
        for (CompletionSuggestion.Entry.Option option : fullPinyinSuggestion.getOptions()) {
            Map<String, Object> sourceMap = option.getHit().getSourceAsMap();//获取文档
            Map<String, Object> chineseText = (Map<String, Object>) sourceMap.get("chinese");//获取文档中的中文
            String sugChinese = (String) chineseText.get("input");
            if (!sugSet.contains(sugChinese)) {
                sugList.add(sugChinese);
                sugSet.add(sugChinese);
            }
        }

        CompletionSuggestion headPinyinSuggestion = suggest.getSuggestion(headPinyinSugName);//获取suggest结果
        //遍历拼音首字母suggest结果
        for (CompletionSuggestion.Entry.Option option : headPinyinSuggestion.getOptions()) {
            Map<String, Object> sourceMap = option.getHit().getSourceAsMap();//获取文档
            Map<String, Object> chineseText = (Map<String, Object>) sourceMap.get("chinese");//获取文档中的中文
            String sugChinese = (String) chineseText.get("input");
            if (!sugSet.contains(sugChinese)) {
                sugList.add(sugChinese);
                sugSet.add(sugChinese);
            }
        }
        return Result.builder()
                .code(HttpStatus.OK.getCode())
                .data(sugList).build();
    }

    @PostMapping("/getsuggest")
    public Result<Object> getSuggest(@RequestBody HotelData hotelData) {
        SearchRequest request = new SearchRequest(INDEX_NAME);
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        List<AggregationBuilder> list = getAgg(hotelData);
        list.stream().forEach(item -> {
            sourceBuilder.aggregation(item);
        });
        request.source(sourceBuilder);

        HotelData resData = HotelData.builder().build();

        log.info("Query DSL : {}", sourceBuilder.toString());
        RestHighLevelClient client = elasticUtil.getHighLevelClient();
        try {
            SearchResponse response = client.search(request, RequestOptions.DEFAULT);

            Aggregations responseAggregations = response.getAggregations();
            Stats priceStats = responseAggregations.get(PRICE_AGG);
            PriceAgg priceAgg = new PriceAgg();
            priceAgg.setAvg(priceStats.getAvg());
            priceAgg.setMax(priceStats.getMax());
            priceAgg.setMin(priceStats.getMin());
            resData.setPriceAgg(priceAgg);

            Map<String, Object> starsMap = new HashMap<>();
            Terms terms = responseAggregations.get(STAR_AGG);
            List<? extends Terms.Bucket> buckets = terms.getBuckets();
            buckets.forEach(item -> {
                String key = item.getKeyAsString();
                Long count = item.getDocCount();
                starsMap.put(key, count);
            });
            resData.setStarsMap(starsMap);

            Map<String, Object> geoMap = new HashMap<>();
            ParsedGeoDistance geoDistance = responseAggregations.get(GEO_AGG);
            if (geoDistance != null) {
                List<? extends Range.Bucket> distanceBuckets = geoDistance.getBuckets();
                distanceBuckets.forEach(item -> {
                    String key = item.getKeyAsString();
                    geoMap.put(key, item.getDocCount());
                });
            }
            resData.setGeoMap(geoMap);
        } catch (IOException e) {
            log.error(e.getMessage(), e);
        }
        return Result.builder()
                .code(HttpStatus.OK.getCode())
                .data(resData).build();
    }

    /**
     * 获取聚合
     *
     * @param hotelData
     * @return
     */
    public List<AggregationBuilder> getAgg(HotelData hotelData) {
        List<AggregationBuilder> builderList = new ArrayList<>();
        builderList.add(AggregationBuilders.stats(PRICE_AGG).field("price"));
        builderList.add(AggregationBuilders.terms(STAR_AGG).field("star"));

        GeoDistanceAggregationBuilder geoDistance = AggregationBuilders.geoDistance(GEO_AGG, new GeoPoint(hotelData.getQuery().getLon(), hotelData.getQuery().getLat()));
        geoDistance.unit(DistanceUnit.KILOMETERS);
        geoDistance.field("location");
        geoDistance.addRange(new GeoDistanceAggregationBuilder.Range("0-3", 0d, 3d));
        geoDistance.addRange(new GeoDistanceAggregationBuilder.Range("0-10", 0d, 10d));
        geoDistance.addRange(new GeoDistanceAggregationBuilder.Range("0-50", 0d, 50d));
        builderList.add(geoDistance);
        return builderList;
    }

    /**
     * 获取BoolQueryBuilder
     * @return
     */
    public BoolQueryBuilder getFilter(HotelData hotelData) {
        BoolQueryBuilder builder = new BoolQueryBuilder();
        HotelDto query = hotelData.getQuery();
        if (query != null) {
            if (StrUtil.isNotBlank(query.getCity())) {
                builder.filter(QueryBuilders.termQuery("city", query.getCity()));
            }
            if (StrUtil.isNotBlank(query.getLocation())) {
                if (StrUtil.isNotBlank(query.getDistance())) {
                    builder.must().add(QueryBuilders.geoDistanceQuery("location")
                            .distance(Double.parseDouble(query.getDistance()), DistanceUnit.KILOMETERS).point(query.getLat(), query.getLon()));
                } else {
                    builder.must().add(QueryBuilders.geoDistanceQuery("location")
                            .distance(50, DistanceUnit.KILOMETERS).point(query.getLat(), query.getLon()));
                }
            }
            if (StrUtil.isNotBlank(query.getSuggestValue())) {
                log.info("关键字查询：" + query.getSuggestValue());
                // 将关键字内容写入kafka
                kafkaUtil.sendMsg(topicName, query.getSuggestValue());
                builder.must().add(QueryBuilders.matchQuery("title", query.getSuggestValue()));
            }
            String price = query.getPrice();
            if (StrUtil.isNotBlank(price)) {
                if (price.contains("-")) {
                    String[] priceStrs = price.split("-");
                    builder.must().add(QueryBuilders.rangeQuery("price")
                            .gte(Double.parseDouble(priceStrs[0]))
                            .lte(Double.parseDouble(priceStrs[1]))
                    );
                }
            }
            String star = query.getStar();
            if (StrUtil.isNotBlank(star)) {
                builder.must().add(QueryBuilders.termQuery("star", star));
            }
        }
        return builder;
    }

    /**
     * 综合排序脚本
     * @param hotelData
     * @return
     */
    public Script getScript(HotelData hotelData) {
        String code = "double score=0;\n" +
                "double titleMatch = _score;\n" +
                "double distanceMatch = decayGeoGauss(params['origin'], params['scale'], params['offset'], params['decay'], doc['location'].value);\n" +
                "double fullRoomMatch = 0;\n" +
                "if(doc['full_room'].value == true) {\n" +
                "    fullRoomMatch = 0.1;\n" +
                "} else {\n" +
                "    fullRoomMatch = 1;\n" +
                "}\n" +
                "double favourableMatch = doc['favourable_percent'].value * 0.01;\n" +
                "score = params['titleWeight'] * titleMatch + params['fullRoomWeight'] * fullRoomMatch + params['distanceWeight'] * distanceMatch + params['favourableWeight'] * favourableMatch;\n" +
                "return score;";
        Map<String, Object> params = new HashMap<>();
        params.put("titleWeight", 30);
        params.put("fullRoomWeight", 50);
        params.put("distanceWeight", 30);
        params.put("favourableWeight", 20);
        params.put("scale", "5km");
        params.put("offset", "0.5km");
        HotelDto hotelDto = hotelData.getQuery();
        params.put("origin", hotelDto.getLon() + "," + hotelDto.getLat());
        params.put("decay", 0.5);
        return new Script(ScriptType.INLINE, Script.DEFAULT_SCRIPT_LANG , code, params);
    }
}
